package com.chat.server.service;

import com.chat.server.ChatServiceGrpc;
import com.chat.server.ChatServiceProto;
import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.stub.StreamObserver;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 服务端，重写ChatServiceImplBase
 * 1、重写StreamObserver<ChatRequest> chatHello(StreamObserver<ChatReply> responseObserver)方法，
 *    取消息内容：在chathello方法中，实现StreamObserver<V> 方法，客户端消息在void onNext(V var1); V参数可以取到
 *    写消息内容：responseObserver.onNext()方法
 */
public class ChatServer extends ChatServiceGrpc.ChatServiceImplBase {
    private static final List<StreamObserver<ChatServiceProto.ChatReply>> clientList = new LinkedList<>();
    private static Server server;
    int port = 50051;

    //服务端启动
    public static void main(String[] args) {
        ChatServer chatService = new ChatServer();
        try {
            chatService.start();
            chatService.blockUntilShutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //服务的启动方法
    public void start() throws IOException {
        server = ServerBuilder.forPort(port)
                .addService(new ChatServer())
                .build()
                .start();
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            // Use stderr here since the logger may have been reset by its JVM shutdown hook.
            System.err.println("*** shutting down gRPC server since JVM is shutting down");
            try {
                System.out.println("服务端异常调用stop方法");
                ChatServer.this.stop();
            } catch (InterruptedException e) {
                System.out.println("服务端异常调用stop方法异常");
                e.printStackTrace(System.err);
            }
            System.err.println("*** server shut down");
        }));
        System.out.println("Server Start Successfully!");
    }


    @Override
    public StreamObserver<ChatServiceProto.ChatRequest> chatHello(StreamObserver<ChatServiceProto.ChatReply> responseObserver) {

        //StreamObserver<ChatServiceProto.ChatRequest> 是一个接口，这里new一个接口，并实现接口
        StreamObserver<ChatServiceProto.ChatRequest> streamObserver = new StreamObserver<ChatServiceProto.ChatRequest>() {
            @Override
            public void onNext(ChatServiceProto.ChatRequest chatRequest) {
                System.out.println("[收到客户端消息]: " + chatRequest.getName());
                responseObserver.onNext(ChatServiceProto.ChatReply.newBuilder().setMessage("hello client ,I'm Java grpc Server,your message '" + chatRequest.getName()  + "'").build());
                // 添加在线客户端
                System.out.println("客户端+1");
                clientList.add(responseObserver);
                // 开启定时器，向客户端发送消息
                ScheduledExecutorService timer = Executors.newScheduledThreadPool(1);
                timer.scheduleWithFixedDelay(() -> {
                    //lamba表达式写一个线程
                    try {
                        responseObserver.onNext(ChatServiceProto.ChatReply.newBuilder().setMessage(System.nanoTime() + " → 您好，我是服务端!").build());
                        System.out.println("服务端给客户端发送消息："+System.nanoTime() + " → 您好，我是服务端!");
                    } catch (Exception e) {
                        // 移除客户端
                        System.out.println("移除客户端"+",当前客户端数量："+clientList.size());
                        clientList.remove(responseObserver);
                        System.out.println("移除客户端后"+",当前客户端数量："+clientList.size());
                        timer.shutdownNow();
                    }
                }, 0, 3, TimeUnit.SECONDS);
            }

            @Override
            public void onError(Throwable throwable) {
                throwable.fillInStackTrace();
            }

            @Override
            public void onCompleted() {
                responseObserver.onCompleted();
            }
        };
        return streamObserver;
    }


    //停止服务
    public void stop() throws InterruptedException {
        if (server != null) {
//            server.shutdown();
            server.shutdown().awaitTermination(30, TimeUnit.SECONDS);
        }
    }

    /**
     * Await termination on the main thread since the grpc library uses daemon threads.
     */
    public void blockUntilShutdown() throws InterruptedException {
        if (server != null) {
            server.awaitTermination();
        }
    }

}
